package com.huhong.demo.apps

import akka.actor.{ActorPath, ActorSystem, Props, _}
import com.huhong.demo.actors.SimpleClusterListener
import com.typesafe.config.ConfigFactory
import akka.cluster.client.{ClusterClient, ClusterClientReceptionist, ClusterClientSettings}
import com.typesafe.config.ConfigFactory

import akka.cluster.Cluster
import akka.cluster.ClusterEvent._


object DemoMaster {

  def main(args: Array[String]): Unit = {
    val config = ConfigFactory.parseString(
      """
     akka {
       actor {
         provider = "akka.cluster.ClusterActorRefProvider"
       }
       remote {
         transport = "akka.remote.netty.NettyRemoteTransport"
         log-remote-lifecycle-events = off
         netty.tcp {
           hostname = "127.0.0.1"
           port = 2551
         }
       }

       cluster {
         seed-nodes = [
           "akka.tcp://ClusterSystem@127.0.0.1:2551"
           ]

         roles = [master]

         auto-down = on
       }
     }""")

    val system = ActorSystem("ClusterSystem", ConfigFactory.load(config))
    val master = system.actorOf(Props[ClusterMaster], "master")
    ClusterClientReceptionist(system).registerService(master)

  }

  class ClusterMaster extends Actor with ActorLogging {

    val cluster = Cluster(context.system)

    override def preStart(): Unit = {
      //#subscribe
      cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
        classOf[MemberEvent], classOf[UnreachableMember])
      //#subscribe
    }

    def receive = {
      case MemberUp(member) =>
        log.info("Member is Up: {}", member.address)
      case UnreachableMember(member) =>
        log.info("Member detected as unreachable: {}", member)
      case MemberRemoved(member, previousStatus) =>
        log.info("Member is Removed: {} after {}",
          member.address, previousStatus)
      case _: MemberEvent => // ignore

      case e =>
        log.info(s"from master : $e : $sender")
        sender ! "master : how are you?"
    }
  }

}

object DemoMember extends App {

  start(3002)

  def start(port: Int) {
    val config = ConfigFactory.parseString(
      s"""
     akka {
       actor {
         provider = "akka.cluster.ClusterActorRefProvider"
       }

       remote {
         transport = "akka.remote.netty.NettyRemoteTransport"
         log-remote-lifecycle-events = off
         netty.tcp {
          hostname = "127.0.0.1"
          port = ${port}
         }
       }

       cluster {
         seed-nodes = [
           "akka.tcp://ClusterSystem@127.0.0.1:2551"
           ]

         auto-down = on
       }
     }""")

    val system = ActorSystem("ClusterSystem", ConfigFactory.load(config))
    val clusterMember = system.actorOf(Props[ClusterMember], "member")
    ClusterClientReceptionist(system).registerService(clusterMember)
  }


  class ClusterMember extends Actor with ActorLogging {
    def receive = {
      case e =>
        log.info(s"from member : $e : $sender")
        sender ! "member : how are you?"
    }
  }

}

object DemoClient {

  def main(args: Array[String]) {
    val config = ConfigFactory.parseString(
      """
     akka {
       actor {
         provider = "akka.remote.RemoteActorRefProvider"
       }

       remote {
         transport = "akka.remote.netty.NettyRemoteTransport"
         log-remote-lifecycle-events = off
         netty.tcp {
          hostname = "127.0.0.1"
          port = 0
         }
       }
     }""")

    val system = ActorSystem("OTHERSYSTEM", ConfigFactory.load(config))
    val initialContacts = Set(
      ActorPath.fromString("akka.tcp://ClusterSystem@127.0.0.1:2551/system/receptionist"),
      ActorPath.fromString("akka.tcp://ClusterSystem@127.0.0.1:3000/system/receptionist"), ActorPath.fromString("akka.tcp://ClusterSystem@127.0.0.1:3001/system/receptionist"), ActorPath.fromString("akka.tcp://ClusterSystem@127.0.0.1:3002/system/receptionist"))
    val settings = ClusterClientSettings(system).withInitialContacts(initialContacts)

    val c = system.actorOf(ClusterClient.props(settings), "os-client")

    (1 to 1000).map { i =>
      //      c ! ClusterClient.Send("/user/master", s"hello - $i", localAffinity = true)
      //      c ! ClusterClient.Send("/user/member", s"hello - $i", localAffinity = true)
      c ! ClusterClient.Send("/user/member", s"hello14 - $i", true)
      Thread.sleep(1000)
    }
  }
}


//package spikes.cluster
//
//import akka.actor._
//import com.typesafe.config.ConfigFactory
//import akka.contrib.pattern.{ClusterClient, ClusterReceptionistExtension}
//
//object DemoMaster {
//
//  def main(args: Array[String]): Unit = {
//    val config = ConfigFactory.parseString("""
//     akka {
//       actor {
//         provider = "akka.cluster.ClusterActorRefProvider"
//       }
//       remote {
//         transport = "akka.remote.netty.NettyRemoteTransport"
//         log-remote-lifecycle-events = off
//         netty.tcp {
//           hostname = "127.0.0.1"
//           port = 2551
//         }
//       }
//
//       cluster {
//         seed-nodes = [
//           "akka.tcp://ClusterSystem@127.0.0.1:2551"
//           ]
//
//         roles = [master]
//
//         auto-down = on
//       }
//     }""")
//
//    val system = ActorSystem("ClusterSystem", ConfigFactory.load(config))
//    val master = system.actorOf(Props[ClusterMaster], "master")
//    ClusterReceptionistExtension(system).registerService(master)
//  }
//
//  class ClusterMaster extends Actor with ActorLogging {
//    def receive = {
//      case e =>
//        log.info(s"from master : $e : $sender")
//        sender ! "master : how are you?"
//    }
//  }
//}
//
//object DemoMember {
//
//  def main(args: Array[String]) {
//    val config = ConfigFactory.parseString("""
//     akka {
//       actor {
//         provider = "akka.cluster.ClusterActorRefProvider"
//       }
//
//       remote {
//         transport = "akka.remote.netty.NettyRemoteTransport"
//         log-remote-lifecycle-events = off
//         netty.tcp {
//          hostname = "127.0.0.1"
//          port = 3000
//         }
//       }
//
//       cluster {
//         seed-nodes = [
//           "akka.tcp://ClusterSystem@127.0.0.1:2551"
//           ]
//
//         auto-down = on
//       }
//     }""")
//
//    val system = ActorSystem("ClusterSystem", ConfigFactory.load(config))
//    val clusterMember = system.actorOf(Props[ClusterMember], "member")
//    ClusterReceptionistExtension(system).registerService(clusterMember)
//  }
//
//  class ClusterMember extends Actor with ActorLogging {
//    def receive = {
//      case e =>
//        log.info(s"from member : $e : $sender")
//        sender ! "member : how are you?"
//    }
//  }
//}
//
//object DemoClient {
//
//  def main(args : Array[String]) {
//    val config = ConfigFactory.parseString("""
//     akka {
//       actor {
//         provider = "akka.remote.RemoteActorRefProvider"
//       }
//
//       remote {
//         transport = "akka.remote.netty.NettyRemoteTransport"
//         log-remote-lifecycle-events = off
//         netty.tcp {
//          hostname = "127.0.0.1"
//          port = 5000
//         }
//       }
//     }""")
//
//    val system = ActorSystem("OTHERSYSTEM", ConfigFactory.load(config))
//    val initialContacts = Set(
//      system.actorSelection("akka.tcp://ClusterSystem@127.0.0.1:2551/user/receptionist"),
//      system.actorSelection("akka.tcp://ClusterSystem@127.0.0.1:3000/user/receptionist"))
//
//    val c = system.actorOf(ClusterClient.props(initialContacts), "os-client")
//
//    (1 to 1000).map { i =>
//      c ! ClusterClient.Send("/user/master", s"hello - $i", localAffinity = true)
//      c ! ClusterClient.Send("/user/member", s"hello - $i", localAffinity = true)
//
//      Thread.sleep(1000)
//    }
//  }
//}